/*
 * Copyright 2008-2009 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.hasor.registry.server.pushing;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import net.hasor.core.AppContext;
import net.hasor.core.Init;
import net.hasor.core.Inject;
import net.hasor.core.Singleton;
import net.hasor.registry.server.domain.LogUtils;
import net.hasor.registry.server.manager.ServerSettings;
import net.hasor.registry.trace.TraceUtil;
import net.hasor.rsf.InterAddress;
import net.hasor.rsf.RsfContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 推送服务触发器
 *
 * @author 赵永春(zyc@hasor.net)
 * @version : 2016年3月1日
 */
// TODO: 17/5/2 by zmyer
@Singleton
public class PushQueue implements Runnable {
    protected Logger logger = LoggerFactory.getLogger(getClass());
    //数据队列对象
    private LinkedBlockingQueue<PushEvent> dataQueue;
    //数据推动线程
    private ArrayList<Thread> threadPushQueue;
    //处理器映射表
    private Map<RsfCenterEventEnum, PushProcessor> processorMapping;
    //rsf上下文对下那个
    @Inject
    private RsfContext rsfContext;
    //rsf注册中心配置
    @Inject
    private ServerSettings rsfCenterCfg;

    // TODO: 17/5/7 by zmyer
    @Init
    public void init() {
        //读取app上下文对象
        AppContext app = this.rsfContext.getAppContext();
        //创建处理器映射表
        this.processorMapping = new HashMap<RsfCenterEventEnum, PushProcessor>();
        for (RsfCenterEventEnum eventType : RsfCenterEventEnum.values()) {
            //根据事件类型,查找对应的处理器
            PushProcessor processor = app.getInstance(eventType.getProcessorType());
            //将处理器与事件类型的关系插入到映射表中
            this.processorMapping.put(eventType, processor);
            logger.info("pushQueue processor mapping {} -> {}", eventType.forCenterEvent(), eventType.getProcessorType());
        }

        //创建事件队列
        this.dataQueue = new LinkedBlockingQueue<PushEvent>();
        //创建线程推送队列
        this.threadPushQueue = new ArrayList<Thread>();
        //读取线程数目
        int threadSize = rsfCenterCfg.getThreadSize();
        for (int i = 1; i <= threadSize; i++) {
            //创建线程
            Thread pushQueue = new Thread(this);
            pushQueue.setDaemon(true);
            pushQueue.setName("Rsf-Center-PushQueue-" + i);
            pushQueue.setContextClassLoader(this.rsfContext.getClassLoader());
            //启动线程
            pushQueue.start();
            //将线程插入到线程队列中
            this.threadPushQueue.add(pushQueue);
        }
        logger.info("PushQueue Thread start.");
    }

    // TODO: 17/5/7 by zmyer
    public void run() {
        while (true) {
            try {
                //推送事件
                PushEvent pushEvent = null;
                //依次从推送事件队列中读取每个推送的事件
                while ((pushEvent = this.dataQueue.take()) != null) {
                    //开始处理推送事件
                    doPush(pushEvent);
                }
            } catch (Throwable e) {
                logger.error(LogUtils.create("ERROR_300_00004")//
                    .addLog("traceID", TraceUtil.getTraceID())//
                    .logException(e).toJson());
            }
        }
    }

    // TODO: 17/5/7 by zmyer
    // - 立刻执行消息推送,返回推送失败的地址列表。
    private List<InterAddress> doPush(PushEvent pushEvent) {
        //从处理映射表中读取对应的处理器对象
        PushProcessor pushProcessor = this.processorMapping.get(pushEvent.getPushEventType());
        if (pushProcessor != null) {
            //开始使用处理器处理对应的事件
            return pushProcessor.doProcessor(pushEvent);
        } else {
            logger.error(LogUtils.create("ERROR_300_00005")//
                .addLog("traceID", TraceUtil.getTraceID())//
                .addLog("pushEventType", pushEvent.getPushEventType().name())//
                .toJson());
        }
        //返回推动地址
        return pushEvent.getTarget();
    }

    // TODO: 17/5/7 by zmyer
    // - 将消息推送交给推送线程,执行异步推送。
    public boolean doPushEvent(PushEvent eventData) {
        //如果推送队列中的事件数目超过了最大限制
        if (this.dataQueue.size() > this.rsfCenterCfg.getQueueMaxSize()) {
            try {
                //直接睡眠等待
                Thread.sleep(this.rsfCenterCfg.getSleepTime());
            } catch (Exception e) {
                logger.error(LogUtils.create("ERROR_300_00004")//
                    .addLog("traceID", TraceUtil.getTraceID())//
                    .logException(e).toJson());
            }
            //如果依旧超过了限制,则直接返回不能再插入的状态
            if (this.dataQueue.size() > this.rsfCenterCfg.getQueueMaxSize()) {
                return false;//资源还是紧张,返回 失败
            }
        }
        //将事件插入到队列中
        this.dataQueue.offer(eventData);//资源不在紧张，加入到队列中。
        return true;
    }
}